package com.ruibang.glass.job.service;

import com.teaming.cloud.framework2.elastic.job3.annotation.ElasticJobScheduler;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.stereotype.Service;

import java.io.*;
import java.net.InetAddress;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;


/**
 * 基于注解形式script job
 */
//@Service
//@ElasticJobScheduler(cron = "30 */10 * * * ?", shardingTotalCount = 1, name = "RTDataElasticJob",
//        shardingItemParameters = "0=0", jobParameters = "parameter")
@Slf4j
public class RTDataElasticJob implements SimpleJob {

    @Override
    public void execute(ShardingContext shardingContext) {
        try {
            log.error(String.format("RTDataElasticJob starting..., IP: %s,Thread ID: %s", InetAddress.getLocalHost().getHostAddress(),
                    Thread.currentThread().getId()));
            StringBuilder msgBuilder = new StringBuilder();
            String[] command = new String[]{"python", "/soft/pyscript/main.py", "-t", "rtd"};
            ProcessBuilder processBuilder = new ProcessBuilder(command);
            processBuilder.directory(new File("/soft/pyscript"));
            // 重定向输入输出
            processBuilder.redirectErrorStream(true);
            processBuilder.redirectOutput(ProcessBuilder.Redirect.appendTo(new File("/soft/pyscript/log/rtd.log")));
            ExecutorService executor = Executors.newSingleThreadExecutor();
            Future<Process> future = executor.submit(new Callable<Process>() {
                @Override
                public Process call() throws Exception {
                    return processBuilder.start();
                }
            });
            try {
                Process process = future.get(5400, TimeUnit.SECONDS); // 设置超时时间为9分钟
                // 启动线程读取输出流
                BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                new Thread(() -> {
                    String line;
                    try {
                        while ((line = reader.readLine()) != null) {
                            log.info(line); // 处理输出
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }).start();
                int exitCode = process.waitFor();
                if (reader != null) {
                    reader.close();
                }
                if (process != null) {
                    process.destroyForcibly();
                }
                log.info("RTDataJob Process exitCode: " + exitCode);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("RTDataJob Process timed out or error occurred: " + e);
                // 终止进程
                future.cancel(true);
            } finally {
                executor.shutdownNow();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        log.error("RTDataElasticJob execute successful.");
    }

}
